package com.csw.flink.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object Demo04KafkaSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //创建配置文件对象
    val properties = new Properties()
    properties.setProperty("bootstrap.servers","master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id","csw")
    properties.setProperty("auto.offset.reset","earliest")//earliest：读取所有数据 latest：读取最新数据

    /**
      * 读取数据的位置：
      * 1、consumer.setStartFromEarliest() 从最早开始读取数据
      * 2、consumer.setStartFromLatest() 从最新读取数据
      * 3、consumer.setStartFromGroupOffsets()  (默认值)
      *    当消费者不存在时默认读取最新的数据
      *    当消费者不存在时  properties.setProperty("auto.offset.reset","earliest") 从最早读取数据
      *
      *    当消费者已经存在的时候 ，从上一次读取的位置接着读取数据，  保证同一条数据在一个组内只读取一次
      *
      *消费者消费偏移量提交时间默认是5秒提交一次(kafka的消费偏移量保存在kafka的一个topic中)
      *
      */

    /**
      * 创建kafka消费者
      */
    val consumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("test_topicl",new SimpleStringSchema(),properties)

    //读取所有数据
//    consumer.setStartFromEarliest()// start from the earliest record possible
    //读取最新的数据
//    consumer.setStartFromLatest()//start from the latest record
    //指定时间戳读取数据
//    consumer.setStartFromTimestamp(...)//start from specified epoch timestamp （milliseconds）
    //根据消费组读取数据，每一条数据在一个组内只读取一次（偏移量没提交时可能重复读取数据）
    consumer.setStartFromGroupOffsets()//the default behaviour



    val kafkaDS: DataStream[String] = env.addSource(consumer)

    val countDS: DataStream[(String, Int)] = kafkaDS.flatMap(_.split(",")).map((_,1)).keyBy(_._1).sum(1)


    countDS.print()

    env.execute()
  }
}
